package com.surfilter.massdata.spark.task.etlnew;

import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.storage.StorageLevel;

import scala.Tuple2;

import com.act.sparkanalyz.log.SysLog;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.surfilter.massdata.spark.bean.CipTopDomain;
import com.surfilter.massdata.spark.bean.DayDomain;
import com.surfilter.massdata.spark.bean.ExDomain;
import com.surfilter.massdata.spark.bean.ExDomainRegex;
import com.surfilter.massdata.spark.bean.TopDomain;
import com.surfilter.massdata.spark.model.DnsStatConsts;
import com.surfilter.massdata.spark.model.StructDomain;
import com.surfilter.massdata.spark.model.StructIP;
import com.surfilter.massdata.spark.util.BaseRedisDao;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.DomainDatabase;
import com.surfilter.massdata.spark.util.IPDatabase;
import com.surfilter.massdata.spark.util.StringUtil;
import com.surfilter.massdata.spark.util.URLUtil;

public class EtlDayTaskNew {
	public static String path = ConfigUtil.get("dnsfile_path", "");
	public static String[] DIps = ConfigUtil.getArray("surfilter.114.ips");
	public static String[] CIps = ConfigUtil.getArray("surfilter.filter.cips");
	
	public static String URL_PREFIX_HTTP = "http://";
	public static String DOMIAIN_REGX = "^([a-zA-Z0-9]([a-zA-Z0-9\\-]{0,61}[a-zA-Z0-9])?\\.)+[a-zA-Z]{2,6}$";
	public static Pattern domainPattern = Pattern.compile(DOMIAIN_REGX);

	public static void main(String[] args) {
		long start=System.currentTimeMillis();
		long temp=start;
		long temp_end=start;
		double exectime=0.0;
		String ymdhms = "yyyy-MM-dd HH:mm:ss";
		SimpleDateFormat yyyyMMddHHmmss = new SimpleDateFormat(ymdhms);
		System.out.println("etl begin:"+yyyyMMddHHmmss.format(new Date())+"........................");
		Set<String> filterDIps =new HashSet<String>();
		Set<String> filterCIps =new HashSet<String>();
		String dayStr = "";
		Date date = null;
		String dns_src_hivedb = ConfigUtil.get("dns_src_hivedb", "default");
		String dns_src_table = ConfigUtil.get("dns_src_table", "dns");
		String dns_src_table_day = ConfigUtil.get("dns_src_table_day", "ds");
		String hiveDB = ConfigUtil.get("hiveDB", "dns_smcs");
		String day_domain = ConfigUtil.get("domain_etl_table", "day_domain");
		System.out.println("day_domain:"+day_domain);
		try {
			for (String ip : DIps) {
				filterDIps.add(ip) ;
			}
			for (String ip : CIps) {
				filterCIps.add(ip);
			}
			System.out.println("filterDIps:"+filterDIps.toString());
			System.out.println("filterCIps:"+filterCIps.toString());
			SparkConf conf = new SparkConf().setAppName("etl_task_new");
			JavaSparkContext context = new JavaSparkContext(conf);
			HiveContext hiveContext = new HiveContext(context);
		if (args.length > 0) {
				// 时间格式:2017041230 %Y%m%d%H
				dayStr = args[0];
			} else {
				System.out.println("请输入时间参数，格式：2017041230");
				System.exit(1);
			}
			date = DateUtil.getExecDate(DateUtil.getYesterday(), dayStr);
			String time = DateUtil.getCurrentymd(date);//2017-04-12
			String day = DateUtil.getyyyyMMddHH(date);//20170412
			String hour=DateUtil.getCurrentHour_STR(date);//HH
			int hour_i=DateUtil.getCurrentHour(date);			
			String sql="select host,destip,srcip,count,accesstime  from "
					+ dns_src_hivedb + "." + dns_src_table
					+ " where "+dns_src_table_day+"=" + day + " and hour ="+hour_i;
			System.out.println("sql:"+sql);
	           
			JavaRDD<String> rdd = hiveContext
			.sql(sql).toJavaRDD().flatMap(new dnsFlatMap(filterDIps,filterCIps,domainPattern))
			//.persist(StorageLevel.MEMORY_AND_DISK_SER())
			;
			
			
			
            System.out.println("rdd.getNumPartitions:"+rdd.getNumPartitions());
            //rdd.coalesce(200);
            System.out.println("rdd.getNumPartitions:"+rdd.getNumPartitions());
            //发现新的泛域名
            genExdomain(rdd);
            
            temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("Find ExDomain exectime: "+exectime+" min........................");

            //读取泛域名库
			BaseRedisDao redisDao = new BaseRedisDao();
			Map<String, ExDomainRegex> exDomainMap = redisDao
					.getAllExDomainMap("EXDOMAIN");
			final Broadcast<Map<String, ExDomainRegex>> broadExDomain = context
					.broadcast(exDomainMap);
			
			//泛域名替换
			//输出:domain+"\t"+dip+"\t"+cip+"\t"+count+"\t"+time+"\t"+topDomain
			JavaRDD<String> dns_1=rdd.map(new dnsExdomain(broadExDomain.getValue()));
			
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			
			dns_1.persist(StorageLevel.MEMORY_AND_DISK_SER());
			
			
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			
			//dwb_domain_all< 顶级域名 ，访问量>
			JavaPairRDD<String,Long> topDomainRdd= dns_1.mapToPair(new globalTopDomainMap()).reduceByKey(new globalTopDomainReduce());
			
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			
			//dwb_domain_all --输出
			insertGlobalTopDomain(topDomainRdd,hiveContext,hiveDB,time,hour);
			
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("Write dwb_domain_all exectime: "+exectime+" min........................");
			
			//IPDatabase 广播变量
			final Broadcast<IPDatabase> ipBroadcast = context.broadcast(IPDatabase.getInstance());
			//DomainDatabase 广播变量
			Broadcast<DomainDatabase> domainBroadcast = context.broadcast(DomainDatabase.getInstance());
			//dns_1 DIP省内IP过滤
			JavaPairRDD<String, String> dns_2=dns_1.flatMapToPair(new dipPairFlat(ipBroadcast.getValue()))
					.reduceByKey(new dipReduce()).mapToPair(new dipDomainPair()).reduceByKey(new dipDomainReduce());
			//day_domain --输出
			insertDayDomain(dns_2,hiveContext,hiveDB,time,hour,domainBroadcast);
			
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("Write day_domain exectime: "+exectime+" min........................");
			
			//dwb_domain_cip <cip 顶级域名 ，访问量>
			JavaPairRDD<String,Integer> cipTopDomainRDD=dns_1.flatMapToPair(new cipTopDomainFlatMap(ipBroadcast.getValue()))
					.reduceByKey(new cipTopDomainReduce());
			//dwb_domain_cip --输出
			insertCipTopDomain(cipTopDomainRDD,hiveContext,hiveDB,time,hour,ipBroadcast,date);
			dns_1.unpersist();
			ipBroadcast.destroy();
			broadExDomain.destroy();
			//rdd.unpersist();
			temp_end=System.currentTimeMillis();
            exectime=(temp_end-temp)*1.0/(1000*60);
            temp=temp_end;
			System.out.println("Write dwb_domain_cip exectime: "+exectime+" min........................");
			
			long end=System.currentTimeMillis();
			exectime=(end-start)*1.0/(1000*60);
			System.out.println("etl exectime: "+exectime+" min........................");
			System.out.println("etl end:"+yyyyMMddHHmmss.format(new Date())+"........................");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
    
	private static void genExdomain(JavaRDD<String> rdd){
		try{
			//泛域名处理--开始
			JavaPairRDD<String, String> exd1 = rdd
					.mapToPair(new PairFunction<String, String, Integer>() {
						private static final long serialVersionUID = 1L;

						@Override
						public Tuple2<String, Integer> call(String t)
								throws Exception {
							String [] line =StringUtils.split(t, "\t");
//							String domain = t.getAs("host").toString() ;
//							String dip = t.getAs("destip").toString() ;
							String domain=line[0];
							String dip=line[1];
							String topDomain=line[5];
							return new Tuple2<String, Integer>(domain + "\t"
									+ dip +"\t"+topDomain, 1);
						}

					})
					.reduceByKey(new Function2<Integer, Integer, Integer>() {
						private static final long serialVersionUID = 1L;

						// domain IP v1
						@Override
						public Integer call(Integer v1, Integer v2)
								throws Exception {
							// TODO Auto-generated method stub
							return 1;
						}

					})
					.mapToPair(
							new PairFunction<Tuple2<String, Integer>, String, String>() {

								@Override
								public Tuple2<String, String> call(
										Tuple2<String, Integer> t)
										throws Exception {
									// key:domain dip
									String[] key = StringUtils
											.split(t._1, "\t");
									String topDomain = key[2];
									return new Tuple2<String, String>(key[1]
											+ "\t" + topDomain, "1");
								}

							}).persist(StorageLevel.MEMORY_AND_DISK_SER());
			genExdomain_new(exd1);
			exd1.unpersist();
			//泛域名处理--结束
		}catch(Exception e){
			e.printStackTrace();
		}
		
	}
	private static void genExdomain_new(JavaPairRDD<String, String> exd1) {

		try {
			SysLog.info("*****************genExdomain************************");

			Map<String, Object> exd_count = exd1.countByKey();
			 System.out.println("exd_count.size:"+exd_count.size());
			Iterator it = exd_count.keySet().iterator();
			int judgeNum = Integer.valueOf(ConfigUtil.get(
					"surfilter.massdata.exdomain.judge.num", "200"));
			System.out.println("judgeNum:" + judgeNum);
			Map<String, Object> exdmap = new HashMap<String, Object>();
			// 过滤掉小于阀值的key
			while (it.hasNext()) {
				String key = it.next().toString();
				int num = Integer.valueOf(exd_count.get(key).toString());
				if (num > judgeNum) {
					exdmap.put(key, 1);
				}
			}

			int maxLimit = Integer.valueOf(ConfigUtil.get(
					"surfilter.massdata.exdomain.max.limit", "10000"));
			exd1.filter(new exdFilter_new(exdmap))
					.mapToPair(new transDomainKeyMapToPair_new())
					.reduceByKey(new ipMerge_new())
					.foreachPartition(new FlatMapExDomain_new(maxLimit));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	private static void insertGlobalTopDomain(JavaPairRDD<String,Long> topDomainRdd,HiveContext hiveContext,String hiveDB,String day,String hour){
		JavaRDD<TopDomain> topDomainRDD=topDomainRdd.mapPartitions(new globalTopDomainFlatMap());
		if(topDomainRDD.isEmpty()){
			return;
		}
		DataFrame topDomainDF = hiveContext.createDataFrame(topDomainRDD,	TopDomain.class);
		topDomainDF.registerTempTable("t_domain_all");
		String sql="insert into table "
				+ hiveDB
				+ "."
				+ "dwb_domain_all"
				+ "  partition( ds='"
				+ day
				+ "',hour="+hour+") select top_domain,visit_count from t_domain_all";
		System.out.println("dwb_domain_all insert sql:");
		System.out.println(sql);
		hiveContext.sql(sql);
		hiveContext.dropTempTable("t_domain_all");
	}
	
	private static void insertCipTopDomain(
			JavaPairRDD<String, Integer> cipTopDomainRDD,
			HiveContext hiveContext, String hiveDB, String day, String hour,Broadcast<IPDatabase> ipBroadcast,Date date) {
		JavaRDD<CipTopDomain> cipTopDomainRDD_Res=cipTopDomainRDD.mapPartitions(new cipTopDomainMapPartitions(ipBroadcast.getValue()));
		if(cipTopDomainRDD_Res.isEmpty()){
			return;
		}
		DataFrame cipTopDomainDF = hiveContext.createDataFrame(cipTopDomainRDD_Res,	CipTopDomain.class);
		cipTopDomainDF.registerTempTable("t_domain_cip");
		cipTopDomainDF.persist();
		/*
		String sql="insert into table "
				+ hiveDB
				+ "."
				+ "dwb_domain_cip"
				+ "  partition( ds='"
				+ day
				+ "',hour="+hour+") select cip,top_domain,dip_flag,d_city,city,operators_id,ip_type,user_type,visit_count from t_domain_cip";
				*/
		//生成 hive 表dwb_domain_cip的数据
		String sql="insert into table "
				+ hiveDB
				+ "."
				+ "dwb_domain_cip"
				+ "  partition( ds='"
				+ day
				+ "',hour="+hour+") select cip,city,operators_id,ip_type,user_type, sum(visit_count) visit_count from t_domain_cip group by cip,city,operators_id,ip_type,user_type ";
		System.out.println("dwb_domain_cip insert sql:");
		System.out.println(sql);
		hiveContext.sql(sql);
	
		int year = DateUtil.getCurrentYear(date);
		int half_year = DateUtil.getHalfYear(date);
		int quarter= DateUtil.getQuarter(date);
		int month = DateUtil.getCurrentMonth(date);
		int week = DateUtil.getCurrentWeek(date);
		int day_table = DateUtil.getCurrentDay(date); 

		/*
		String sql_SAN011="insert into table "
				+ hiveDB
				+ "."
				+ "dws_hour_san011"
				+ "  partition( ds='"
				+ day
				+ "',hour_="+hour+") select "+year+", "+half_year+", "+quarter+", "+month+", "+week+", "+day_table+", "+hour
				+ ", 'WEBSITE_COUNT_SUM',"
				+ " dip_flag, sum(visit_count),0 from t_domain_cip group by dip_flag";
		System.out.println("sql_SAN011 insert sql:");
		System.out.println(sql_SAN011);
		hiveContext.sql(sql_SAN011);
		*/
		
		String sql_SAN011_s="insert into table "
				+ hiveDB
				+ "."
				+ "dws_hour_san011"
				+ "  partition( ds='"
				+ day
				+ "',hour_="+hour+") select "+year+", "+half_year+", "+quarter+", "+month+", "+week+", "+day_table+", "+hour
				+ ", 'WEBSITE_COUNT_SUM',"
				+ " dip_flag, visit_count,0 from (select dip_flag, sum(visit_count) as visit_count from  t_domain_cip group by dip_flag) t"
				;
		System.out.println("sql_SAN011_s insert sql:");
		System.out.println(sql_SAN011_s);
		hiveContext.sql(sql_SAN011_s);
		
		
		
		long temp=System.currentTimeMillis();
		
		//
		String sql_1="insert into table "
				+ hiveDB
				+ "."
				+ "dwb_domain_cip_dip"
				+ "  partition( ds='"
				+ day
				+ "',hour="+hour+") select top_domain,d_city,city,operators_id,ip_type,sum(visit_count) as visit_count from t_domain_cip where dip_flag='1' group by top_domain,d_city,city,operators_id,ip_type";
		System.out.println("dwb_domain_cip_dip insert sql:");
		System.out.println(sql_1);
		hiveContext.sql(sql_1);
		long temp_end=System.currentTimeMillis();
		double exectime=(temp_end-temp)*1.0/(1000*60);
	    System.out.println("Write dwb_domain_cip_dip exectime: "+exectime+" min........................");
	    
	    hiveContext.dropTempTable("t_domain_cip");
	    cipTopDomainDF.unpersist();
	    
	    /*
	    temp=System.currentTimeMillis();
	    String sql_2="insert into table "
				+ hiveDB
				+ "."
				+ "dwb_domain_cip_dip"
				+ "  partition( ds='"
				+ day
				+ "',hour="+hour+") select top_domain,dip_city,city,operators_id,ip_type,sum(visit_count) as visit_count from "+hiveDB+".dwb_domain_cip where ds='"+day+"'  and hour='"+hour+"' and dip_flag='1' group by top_domain,dip_city,city,operators_id,ip_type";
		System.out.println("dwb_domain_cip_dip hive insert sql:");
		System.out.println(sql_2);
		hiveContext.sql(sql_2);
		 temp_end=System.currentTimeMillis();
		 exectime=(temp_end-temp)*1.0/(1000*60);
	    System.out.println("Write hive dwb_domain_cip_dip exectime: "+exectime+" min........................");
	    */
	}
	
	private static void insertDayDomain(
			JavaPairRDD<String, String> dipDomainRDD, HiveContext hiveContext,
			String hiveDB, String day, String hour,Broadcast<DomainDatabase> domainBroadcast) {
		String p_linces_key="";
		try {
			p_linces_key = new String(ConfigUtil.get("LICENSE_KEY").getBytes("ISO-8859-1"),"UTF-8");
			System.out.println("p_linces_key:"+p_linces_key);
		} catch (UnsupportedEncodingException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		JavaRDD<DayDomain> dipDomainRDD_Res=dipDomainRDD.mapPartitions(new dipDomainFlatMap(domainBroadcast.getValue(),p_linces_key ));
		if(dipDomainRDD_Res.isEmpty()){
			return;
		}
		DataFrame dipDomainDF = hiveContext.createDataFrame(dipDomainRDD_Res,	DayDomain.class);
		dipDomainDF.registerTempTable("t_day_domain");
		String day_domain = ConfigUtil.get("domain_etl_table", "day_domain");
		String sql="insert into table "
				+ hiveDB
				+ "."
				+ day_domain
				+ "  partition( ds='"
				+ day
				+ "',hour="+hour+") select domain,dip,dns_visit_count,findtime,dip_visit_count,top_domain,license,parent_industry_type,suffix from t_day_domain";
		System.out.println("t_day_domain insert sql:");
		System.out.println(sql);
		hiveContext.sql(sql);
		hiveContext.dropTempTable("t_day_domain");

	}
	
	private static void insertCipDipTopDomain(
			JavaPairRDD<String, Long> cipDipTopDomainRDD,
			HiveContext hiveContext, String hiveDB, String day, String hour) {
		JavaRDD<TopDomain> cipDipTopDomainRDD_Res=cipDipTopDomainRDD.mapPartitions(new globalTopDomainFlatMap());
		if(cipDipTopDomainRDD_Res.isEmpty()){
			return;
		}
		DataFrame cipDipTopDomainDF = hiveContext.createDataFrame(cipDipTopDomainRDD_Res,	TopDomain.class);
		cipDipTopDomainDF.registerTempTable("t_domain_cip_dip");
		String sql="insert into table "
				+ hiveDB
				+ "."
				+ "dwb_domain_cip_dip"
				+ "  partition( ds='"
				+ day
				+ "',hour="+hour+") select top_domain,visit_count from t_domain_cip_dip";
		System.out.println("t_domain_cip_dip insert sql:");
		System.out.println(sql);
		hiveContext.sql(sql);
		hiveContext.dropTempTable("t_domain_cip_dip");
	}

}


class exdFilter_new implements Function<Tuple2<String, String>, Boolean> {
	private static final long serialVersionUID = 1L;
	Map<String, Object> exd_count = null;

	public exdFilter_new(Map<String, Object> exd_count) {
		this.exd_count = exd_count;
	}

	@Override
	public Boolean call(Tuple2<String, String> v1) throws Exception {
		String key = v1._1();
		if (exd_count.containsKey(key)) {
			return true;
		} else {
			return false;
		}
	}

}

class FlatMapExDomain_new implements
		VoidFunction<Iterator<Tuple2<String, String>>> {
	private static final long serialVersionUID = 1L;
	int maxLimit = 10000;

	FlatMapExDomain_new(int maxLimit) {
		this.maxLimit = maxLimit;
	}

	@Override
	public void call(Iterator<Tuple2<String, String>> t) throws Exception {
		String EX_DOMAIN_REDIS_KEY = "EXDOMAIN";
		String EX_DOMAIN_DEFAULT_IP = "0.0.0.0";
		BaseRedisDao redisDao = new BaseRedisDao();
		ObjectMapper mapper = new ObjectMapper();
		ObjectWriter writer = mapper.writer();

		while (t.hasNext()) {
			// key:<domaim> value:<dip1,dip2>
			Tuple2<String, String> tt = t.next();
			String domain = tt._1();
			String value = tt._2();

			ExDomain exDomain = new ExDomain();
			// 当前数据找到的域名、ip
			String[] value_list = StringUtils.split(value, ",");
			// redis 里存的域名、ip
			String redis_value = redisDao.hgetExDomain(EX_DOMAIN_REDIS_KEY,
					domain);
			Set ipSet = new HashSet();
			String ips_new = "";
			if (redis_value != null) {
				ExDomain redis_domain = mapper.readValue(
						StringUtil.toLowerJson(redis_value), ExDomain.class);
				String redis_ips = redis_domain.getIps();
				String[] ip_list = StringUtils.split(redis_ips, ",");
				for (String i : ip_list) {
					ipSet.add(i.trim());
				}
			}
			if (ipSet.contains(EX_DOMAIN_DEFAULT_IP)) {

			} else {
				for (String i : value_list) {
					ipSet.add(i.trim());
				}

				if (ipSet.size() > maxLimit) {
					ips_new = EX_DOMAIN_DEFAULT_IP;
				} else {
					Iterator iterator = ipSet.iterator();
					while (iterator.hasNext()) {
						ips_new += iterator.next();
						ips_new += ",";
					}
					ips_new = ips_new.substring(0, ips_new.length() - 1);
				}

				exDomain.setDomain("www." + domain);
				exDomain.setIps(ips_new);
				exDomain.setSuffix(domain);
				exDomain.setFindtime(new SimpleDateFormat("yyyyMMdd")
						.format(new Date()));
				// System.out.println("redis:" +
				// StringUtil.toUpperJson(writer.writeValueAsString(exDomain)));
				redisDao.hsetExDomain(EX_DOMAIN_REDIS_KEY, domain, StringUtil
						.toUpperJson(writer.writeValueAsString(exDomain)));

			}
		}
	}
}

class transDomainKeyMapToPair_new implements
		PairFunction<Tuple2<String, String>, String, String> {

	@Override
	public Tuple2<String, String> call(Tuple2<String, String> t)
			throws Exception {
		String[] key = StringUtils.split(t._1(), "\t");
		String ip = key[0];
		String domain = key[1];
		return new Tuple2<String, String>(domain, ip);
	}

}

class ipMerge_new implements Function2<String, String, String> {

	@Override
	public String call(String v1, String v2) throws Exception {
		if (v1.contains(v2)) {
			return v1;
		} else if (v2.contains(v1)) {
			return v2;
		} else {
			return v1 + "," + v2;
		}

	}

}

class dnsFlatMap implements FlatMapFunction<Row, String>{
	Set<String> filterDIps;
	Set<String> filterCIps;
	String URL_PREFIX_HTTP = "http://";
	Pattern domainPattern ;
	dnsFlatMap(Set<String> filterDIps,Set<String> filterCIps,Pattern domainPattern){
		this.filterDIps=filterDIps;
    	this.filterCIps=filterCIps;
    	this.domainPattern=domainPattern;
	}
	@Override
	public Iterable<String> call(Row t) throws Exception {
		String domain = t.getAs("host").toString().toLowerCase() ;
		String dip = t.getAs("destip").toString() ;
		String cip = t.getAs("srcip").toString() ;
		String count = t.getAs("count").toString() ;
		String accesstime = t.getAs("accesstime").toString() ;
		List <String> list =new ArrayList<String>();
		 
		// 目的IP 114IP过滤
		if (filterDIps.contains(dip)) {
			return list;
		}
		// 源IP CIP过滤
		if (filterCIps.contains(cip)) {
			return list;
		}
		if (StringUtils.isBlank(domain)
				|| StringUtils.isBlank(dip)
				|| StringUtils.isBlank(cip)) {
			return list;
		}
		// 域名合法性校验
		if (!domainPattern.matcher(domain).matches()) {
			return list;
		}

		String topDomain = URLUtil
				.getDomainName(URL_PREFIX_HTTP + domain);
		if (topDomain == null) {
			return list;
		}
		topDomain = topDomain.toLowerCase();
		String dns=domain+"\t"+dip+"\t"+cip+"\t"+count+"\t"+accesstime+"\t"+topDomain;
		list.add(dns);
		return list;
	}
}

class dnsExdomain implements Function<String, String>{
	Map<String, ExDomainRegex> exDomainMap;
	dnsExdomain(Map<String, ExDomainRegex> exDomainMap){
		this.exDomainMap = exDomainMap;
	}
	@Override
	public String call(String v1) throws Exception {
		String[] line=StringUtils.split(v1,"\t");
		String domain=line[0];
		String dip=line[1];
		String cip=line[2];
		String count=line[3];
		String findtime=line[4];
		String topDomain=line[5];
		String exDomain = domain;
		try {
			// 缺陷【smcs_all_v1.0.0.0】
			domain = exdomainDeal(topDomain, dip, exDomainMap);
		} catch (Exception e) {
			domain = null;
		}
		if (domain == null) {
			domain = exDomain;
			exDomain = null;
		}
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 小写的mm表示的是分钟
		java.util.Date date = sdf.parse(findtime);
		long time = date.getTime() / 1000;
		
		return domain+"\t"+dip+"\t"+cip+"\t"+count+"\t"+time+"\t"+topDomain;
	}
	/**
	 * 过滤掉泛域名-IP,将匹配到的泛域名-IP替换成对应的域名-IP
	 * 
	 * @param domain
	 * @param dips
	 * @return
	 */

	private String exdomainDeal(String topDomain, String dips,
			Map<String, ExDomainRegex> exDomainMap) {
		String EX_DOMAIN_DEFAULT_IP = "0.0.0.0";
		if (StringUtils.isBlank(topDomain) || StringUtils.isBlank(dips)) {
			return null;
		}
		List<String> ipList = new ArrayList<String>();
		//ipList.addAll(Arrays.asList(StringUtils.split(dips, "\\|")));
		String exDomain = null;
		if (exDomainMap.containsKey(topDomain)) {
			ExDomainRegex exDomainRegex = exDomainMap.get(topDomain);
			Set<String> exIps = exDomainRegex.getIpSet();
			if (exIps.contains(EX_DOMAIN_DEFAULT_IP)) {
				exDomain = exDomainRegex.getNewName();
				return exDomain;
			} else if (exIps.contains(dips)) {
			//} else if (intersectionSet(ipList, exIps)) {
				exDomain = exDomainRegex.getNewName();
				return exDomain;
			}
		}
		return exDomain;
	}

//	private boolean intersectionSet(List<String> list, Set<String> setB) {
//		list.retainAll(setB);
//		return list.size() > 0;
//	}
}
class globalTopDomainMap implements PairFunction<String, String, Long>{

	@Override
	public Tuple2<String, Long> call(String t) throws Exception {
		String[] line=StringUtils.split(t,"\t");
		String topdomain=line[5];
		String count=line[3];
		Long count_l=Long.valueOf(count);
		
		return new Tuple2(topdomain, count_l);
	}
	
}
class globalTopDomainReduce implements Function2<Long, Long, Long> {

	@Override
	public Long call(Long v1, Long v2) throws Exception {
		return v1+v2;
	}
	
}

class globalTopDomainFlatMap implements FlatMapFunction<Iterator<Tuple2<String, Long>>, TopDomain>{
	@Override
	public Iterable<TopDomain> call(Iterator<Tuple2<String, Long>> t)
			throws Exception {
		List<TopDomain> list = new ArrayList<TopDomain>();
		while (t.hasNext()) {
			Tuple2<String, Long> tt = t.next();
			TopDomain e = new TopDomain();
			e.setTop_domain(tt._1());
			e.setVisit_count(tt._2());
			list.add(e);
		}
		return list;
	}
}

class cipTopDomainFlatMap implements PairFlatMapFunction<String, String, Integer>{
	IPDatabase ipDatabase;
	cipTopDomainFlatMap(IPDatabase ipDatabase){
		this.ipDatabase=ipDatabase;
	}
	@Override
	public Iterable<Tuple2<String, Integer>> call(String t) throws Exception {
		List<Tuple2<String, Integer>> list =new ArrayList<Tuple2<String, Integer>>();
		String[] line=StringUtils.split(t,"\t");
		String dip=line[1];
		String cip=line[2];
		String count=line[3];
		String topdomain=line[5];
		String dip_flag="2";//1-省内，2-省外
		String d_city=DnsStatConsts.NA_STR;
		//cip省内ip过滤
		StructIP structCip = ipDatabase.get(cip);
		if(structCip != null){
			//dip省内ip过滤
			StructIP structDip = ipDatabase.get(dip);
			if(structDip != null){
				dip_flag="1";
				d_city=structDip.get(DnsStatConsts.IPExField.CITY);
			}
			String key=cip+"\t"+topdomain+"\t"+dip_flag+"\t"+d_city;
			Integer value=Integer.valueOf(count);
			list.add(new Tuple2<String, Integer>(key,value));
		}
		//
		return list;
	}
}

class cipTopDomainReduce implements  Function2<Integer, Integer, Integer>{

	@Override
	public Integer call(Integer v1, Integer v2) throws Exception {
		return v1+v2;
	}
}

class cipTopDomainMapPartitions implements FlatMapFunction<Iterator<Tuple2<String, Integer>>, CipTopDomain>{
	IPDatabase ipDatabase;
	cipTopDomainMapPartitions(IPDatabase ipDatabase){
		this.ipDatabase=ipDatabase;
	}
	@Override
	public Iterable<CipTopDomain> call(Iterator<Tuple2<String, Integer>> t)
			throws Exception {
		List<CipTopDomain> list = new ArrayList<CipTopDomain>();
		while (t.hasNext()) {
			Tuple2<String, Integer> tt = t.next();
			CipTopDomain e = new CipTopDomain();
			String[] key=StringUtils.split(tt._1(),"\t");
			String cip=key[0];
			String top_domain=key[1];
			String dip_flag=key[2];
			String d_city=key[3];
			StructIP structCip = ipDatabase.get(cip);
			if(structCip != null){
				String city = structCip.get(DnsStatConsts.IPExField.CITY);
				String operators_id = structCip.get(DnsStatConsts.IPExField.OPERATORS_ID);
				String iptype=structCip.get(DnsStatConsts.IPExField.IP_TYPE);
				String usertype = structCip.get(DnsStatConsts.IPExField.USING_UNIT_TYPE);
				if(city==null||city.equals("")){
					//N/A
					city=DnsStatConsts.NA_STR;
				}
				if(operators_id==null||operators_id.equals("")){
					//N/A
					operators_id=DnsStatConsts.NA_STR;
				}
				if(iptype==null||iptype.equals("")){
					//N/A
					iptype=DnsStatConsts.NA_STR;
				}
				if(usertype==null||usertype.equals("")){
					//N/A
					usertype=DnsStatConsts.NA_STR;
				}
				e.setCip(cip);
				e.setTop_domain(top_domain);
				e.setDip_flag(dip_flag);
				e.setD_city(d_city);
				e.setCity(city);
				e.setOperators_id(operators_id);
				e.setIp_type(iptype);
				e.setUser_type(usertype);
				e.setVisit_count(tt._2());
				list.add(e);
			}
		}
		return list;
	}
}

class dipDomainFlatMap implements FlatMapFunction<Iterator<Tuple2<String, String>>, DayDomain>{
	DomainDatabase domainData;
	String p_linces_key;
	dipDomainFlatMap(DomainDatabase domainData,String p_linces_key){
		this.domainData=domainData;
		this.p_linces_key=p_linces_key;
	}
	@Override
	public Iterable<DayDomain> call(Iterator<Tuple2<String, String>> t)
			throws Exception {
		String URL_PREFIX_HTTP = "http://";
		Pattern IP_PATTERN = Pattern.compile("(\\d{1,3}\\.){3}(\\d{1,3})");
		List<DayDomain> list = new ArrayList<DayDomain>();
		while (t.hasNext()) {
			Tuple2<String, String> tt = t.next();
			DayDomain e = new DayDomain();
			String domain=tt._1;
			String[] value= StringUtils.split(tt._2,"\t");
			String dip=value[0];
			long dns_visit_count=Long.valueOf(value[1]);
			long findtime=Long.valueOf(value[2]);
			String dip_visit_count="0";
			if(value.length>=4){
				dip_visit_count=value[3];
			}else{
				dip_visit_count=value[1];
			}
			String top_domain=URLUtil
					.getDomainName(URL_PREFIX_HTTP + domain);
			StructDomain structDomain=domainData.get(top_domain);
			if(structDomain != null){
				String licenseKey = structDomain.get(DnsStatConsts.DomainExField.LICENSE_KEY);
				if(licenseKey!=null&&!licenseKey.isEmpty()&&licenseKey.contains(p_linces_key)){
					e.setLicense(1);
				}else{
					e.setLicense(0);
				}
				String type=structDomain.get(DnsStatConsts.DomainExField.PARENT_INDUSTRY_TYPE);
				if(type==null||type.equals("")||type.equalsIgnoreCase("null")){
					e.setParent_industry_type("221");
				}else{
					e.setParent_industry_type(type);
				}
			}else{
				e.setLicense(0);
				e.setParent_industry_type("221");
			}
			e.setDip(dip);
			e.setDns_visit_count(dns_visit_count);
			e.setDomain(domain);
			e.setFindtime(findtime);
			e.setDip_visit_count(dip_visit_count);
			e.setTop_domain(top_domain);
			
			String result = "";
			if(StringUtils.isNotBlank(top_domain)){
				if(IP_PATTERN.matcher(top_domain).matches()){//判断是否为IP
					result =  top_domain;
				}else{
					result = top_domain.substring(top_domain.indexOf(".")+1,top_domain.length());
				}
			}else{
				//表示域名不符合规则
				if(domain.indexOf(".") > 0){
					result = domain.substring(domain.indexOf(".") + 1,domain.length());
				}else{
					result = domain;
				}
			}
			e.setSuffix(result);
			list.add(e);
		}
		return list;
	}
}


class dipPairFlat implements PairFlatMapFunction<String, String, String>{
	IPDatabase ipDatabase;
	dipPairFlat(IPDatabase ipDatabase){
		this.ipDatabase=ipDatabase;
	}
	@Override
	public Iterable<Tuple2<String, String>> call(String t) throws Exception {
		List <Tuple2<String, String>> list =new ArrayList<Tuple2<String, String>>();
 		String[] line=StringUtils.split(t,"\t");
		String dip=line[1];
		StructIP structDip = ipDatabase.get(dip);
		if(structDip != null){
			String domain=line[0];
			String count=line[3];
			String accesstime=line[4];
			String key=domain+"\t"+dip;
			String values=count+"\t"+accesstime;
			list.add(new Tuple2<String, String>(key,values));
		}
		return list;
	}
	
}

class dipReduce implements Function2<String, String, String>{

	@Override
	public String call(String v1, String v2) throws Exception {
		String[] value1=StringUtils.split(v1,"\t");
		String[] value2=StringUtils.split(v2,"\t");
		long count1=Long.valueOf(value1[0]);
		long count2=Long.valueOf(value2[0]);
		long time1=Long.valueOf(value1[1]);
		long time2=Long.valueOf(value2[1]);
		
		long count=count1+count2;
		long time=(time1<time2?time1:time2);
		
		return String.valueOf(count)+"\t"+String.valueOf(time);
	}
}
class dipDomainPair implements PairFunction<Tuple2<String, String>, String, String>{

	@Override
	public Tuple2<String, String> call(Tuple2<String, String> t)
			throws Exception {
		String[] t1=StringUtils.split(t._1,"\t");
		String[] t2=StringUtils.split(t._2,"\t");
		String domain=t1[0];
        String dip=t1[1]; 
        String count=t2[0];
        String time=t2[1];
		return new Tuple2<String, String>(domain,dip+"\t"+count+"\t"+time+"\t"+count);
	}
}

class dipDomainReduce implements  Function2<String, String, String>{

	@Override
	public String call(String v1, String v2) throws Exception {
		//v1:153.35.179.40|221.231.111.52	14	1499065200	8|6
		//v2:153.35.178.160	4	1499065200	4
		System.out.println("v1:"+v1);
		System.out.println("v2:"+v2);
		String[] value_1 = StringUtils.split(v1, "\t");
		String[] value_2 = StringUtils.split(v2, "\t");
		//dip_count
		String[] vl_count=StringUtils.split(value_1[3], "\t");
		String[] v2_count=StringUtils.split(value_2[3], "\t");
		String dip_count=vl_count[0] ;
		if(vl_count.length>1){
			for(int i=1;i<vl_count.length;i++){
				System.out.println("vl_count[i]:"+vl_count[i]);
				dip_count+='|'+vl_count[i];
			}
		}
		
		for(int i=0;i<v2_count.length;i++){
			System.out.println("vl_count[i]:"+vl_count[i]);
			dip_count+='|'+v2_count[i];
		}
		//count
		long count1=Long.valueOf(value_1[1]);
		long count2=Long.valueOf(value_2[1]);
		long count=count1+count2;
		//time
		long time1=0l;
		long time2=0l;
		time1 = Long.valueOf(value_1[2]);
		time2 = Long.valueOf(value_2[2]);
		// 取最早发现时间
		long time = (time1 < time2 ? time1 : time2);
		//dip
		String dip=value_1[0] + "|" + value_2[0];
	    //value
		String value = dip + "\t"
				+ count + "\t"
				+ time+"\t"+dip_count;
		System.out.println("value:"+value);
		return value;
	}
}
